package com.github.mq;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.data.redis.stream.Subscription;

import javax.annotation.PostConstruct;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.time.Duration;
import java.util.List;

@Configuration
@Slf4j
public class RedisStreamConfig {

    @Value("${spring.redis.stream-name}")
    private String streamKey;

    @Value("${spring.redis.consumer-group}")
    private String consumerGroup;

    private final StringRedisTemplate stringRedisTemplate;

    public RedisStreamConfig(StringRedisTemplate stringRedisTemplate) {
        this.stringRedisTemplate = stringRedisTemplate;
    }

    @PostConstruct
    public void initConsumerGroup() {
        try {
            // 检查Stream是否存在
            Boolean streamExists = stringRedisTemplate.hasKey(streamKey);

            if (Boolean.TRUE.equals(streamExists)) {
                // Stream存在，检查消费者组是否存在
                try {
                    StreamInfo.XInfoGroups groups = stringRedisTemplate.opsForStream().groups(streamKey);
                    boolean groupExists = groups.stream()
                            .anyMatch(group -> consumerGroup.equals(group.groupName()));

                    if (!groupExists) {
                        stringRedisTemplate.opsForStream().createGroup(streamKey, ReadOffset.from("0"), consumerGroup);
                        log.info("成功创建消费者组：{}",consumerGroup);
                    } else {
                        log.info("消费者组已存在：{}",consumerGroup);
                    }
                } catch (Exception e) {
                    // 如果获取组信息失败，尝试创建组
                    log.info("获取组信息失败，尝试创建组:{}",e.getMessage());
                }
            } else {
                // Stream不存在，创建Stream和消费者组
                // 使用 MKSTREAM 选项（需要Spring Data Redis 2.6+）
                stringRedisTemplate.opsForStream().createGroup(streamKey, ReadOffset.from("0"), consumerGroup);
                log.info("创建Stream和消费者组:{}",consumerGroup);
            }
        } catch (Exception e) {
            log.info("初始化消费者组失败:{}",e.getMessage());
        }
    }

    @Bean
    public StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> streamMessageListenerContainerOptions() {
        return StreamMessageListenerContainer.StreamMessageListenerContainerOptions
                .builder()
                .pollTimeout(Duration.ofSeconds(1))
                .build();
    }

    @Bean
    public StreamMessageListenerContainer<String, MapRecord<String, String, String>> streamMessageListenerContainer(
            StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> options,
            RedisConnectionFactory redisConnectionFactory,
            StreamListener<String, MapRecord<String, String, String>> streamListener) throws UnknownHostException {

        StreamMessageListenerContainer<String, MapRecord<String, String, String>> container =
                StreamMessageListenerContainer.create(redisConnectionFactory, options);

        String consumerName = InetAddress.getLocalHost().getHostName() + ":" + System.currentTimeMillis();

        // 订阅流
        Subscription subscription = container.receiveAutoAck(
                Consumer.from(consumerGroup, consumerName),
                StreamOffset.create(streamKey, ReadOffset.lastConsumed()),
                streamListener
        );

        container.start();
        System.out.println("Stream监听器已启动，消费者: " + consumerName);
        return container;
    }
}